package com.hdu.dwh.service.impl;

import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
import com.baomidou.mybatisplus.core.metadata.IPage;
import com.baomidou.mybatisplus.extension.plugins.pagination.Page;
import com.hdu.dwh.dtos.TablePageDto;
import com.hdu.dwh.dtos.DataWarehouseTableDto;
import com.hdu.dwh.pojos.PaimonTag;
import com.hdu.dwh.pojos.model_pojos.MetaColumn;
import com.hdu.dwh.pojos.source_access_pojos.DataWarehouseDatabase;
import com.hdu.dwh.pojos.source_access_pojos.DataWarehouseTable;
import com.hdu.dwh.pojos.source_access_pojos.IColumn;
import com.hdu.dwh.pojos.source_access_pojos.TableRelation;
import com.hdu.dwh.mapper.DataWarehouseTableMapper;
import com.hdu.dwh.service.IDatabaseAllService;
import com.hdu.dwh.service.ISourceAccessService;
import com.hdu.dwh.service.ModelService;
import com.hdu.dwh.service.TableRelationService;
import com.hdu.dwh.service.cube_service.CubeService;
import com.hdu.dwh.utils.OperatorUtil;
import com.hdu.dwh.utils.TrinoQueryUtils;
import com.hdu.dwh.service.DataWarehouseDatabaseService;
import com.hdu.dwh.service.DataWarehouseTableService;
import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl;
import com.hdu.dwh.vos.*;
import com.hdu.dwh.vos.dataWareHouse_vos.DataWarehouseTableDetailVo;
import com.hdu.dwh.constants.CommonConstants;
import com.hdu.dwh.constants.ParamConstants;
import com.hdu.dwh.dao.PaimonBaseDao;
import com.hdu.dwh.results.PageResponseResult;
import com.hdu.dwh.results.ResponseResult;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.BeanUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Lazy;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;

import java.sql.Connection;
import java.sql.ResultSet;
import java.sql.SQLException;
import java.util.ArrayList;
import java.util.HashSet;
import java.util.List;
import java.util.Objects;
import java.util.Set;

/**
 * <p>
 * 源数据表表 服务实现类
 * </p>
 *
 * @author bobo
 * @since 2025-04-08
 */
@Slf4j
@Service
@RequiredArgsConstructor
public class DataWarehouseTableServiceImpl extends ServiceImpl<DataWarehouseTableMapper, DataWarehouseTable> implements DataWarehouseTableService {

    private final DataWarehouseTableMapper sourceTableMapper;
    private final IDatabaseAllService databaseAllService;

    @Lazy
    @Autowired
    private DataWarehouseDatabaseService dwhDatabaseService;

    @Autowired
    PaimonBaseDao paimonBaseDao;

    @Autowired
    private TrinoQueryUtils trinoQueryUtils;

    @Autowired
    private TableRelationService tableRelationService;

    @Autowired
    private ISourceAccessService sourceAccessService;

    @Autowired
    @Lazy
    private ModelService modelService;

    @Autowired
    @Lazy
    private CubeService cubeService;

    /**
     * 新增源数据表
     * @param sourceTableDto
     * @return
     */
    @Override
    public ResponseResult add(DataWarehouseTableDto sourceTableDto) {
        if (sourceTableDto == null) {
            return ResponseResult.fail(CommonConstants.PARAM_IS_NULL);
        }
        if(!sourceTableDto.checkParam()){
            return ResponseResult.fail(CommonConstants.PARAM_IS_ERROR);
        }
        // 检查这个数据库下是否存在同表名的表
        if(existsTable(sourceTableDto.getDatabaseId(), sourceTableDto.getTableName())){
            return ResponseResult.fail(CommonConstants.DATAWAREHOUSE_TABLE_EXIST);
        }
        // 检查对应数据库是否存在
        if(!databaseAllService.existsById(sourceTableDto.getDatabaseId())){
            return ResponseResult.fail(CommonConstants.RECORD_NOT_EXIST);
        }
        // 检查对应源表是否存在
        if(sourceTableDto.getSourceTableId()!=null && getById(sourceTableDto.getSourceTableId()) == null){
            return ResponseResult.fail(CommonConstants.RECORD_NOT_EXIST);
        }
        DataWarehouseTable sourceTable = new DataWarehouseTable();
        BeanUtils.copyProperties(sourceTableDto, sourceTable);
        sourceTable.setCreatedBy(OperatorUtil.getCurrentOperator());
        if(!save(sourceTable)){
            return ResponseResult.fail(CommonConstants.ADD_FAILED);
        }
        return ResponseResult.success(CommonConstants.ADD_SUCCESS);
    }

    /**
     * 判断表是否存在
     * @param sourceTableIds
     * @return
     */
    @Override
    public boolean existsByIds(List<Long> sourceTableIds) {
        LambdaQueryWrapper<DataWarehouseTable> queryWrapper = new LambdaQueryWrapper<DataWarehouseTable>()
                .in(DataWarehouseTable::getId, sourceTableIds);
        if(list(queryWrapper).size()!=sourceTableIds.size()){
            return false;
        }
        return true;
    }

    /**
     * 判断是否存在同名表
     * @param databaseId
     * @param tableName
     * @return
     */
    private boolean existsTable(Long databaseId, String tableName) {
        LambdaQueryWrapper<DataWarehouseTable> queryWrapper = new LambdaQueryWrapper<DataWarehouseTable>()
                .eq(DataWarehouseTable::getDatabaseId, databaseId)
                .eq(DataWarehouseTable::getTableName, tableName);
        return exists(queryWrapper);
    }

    /**
     * 查询详细信息
     * @param id
     * @return
     * @throws Exception 
     */
    @Override
    public DataWarehouseTableDetailVo getDetilById(Long id) throws Exception {
        if (id == null) {
            throw new IllegalArgumentException(CommonConstants.PARAM_IS_NULL);
        }
        DataWarehouseTable resTable = getById(id);
        if (resTable == null) {
            throw new IllegalArgumentException(CommonConstants.RECORD_NOT_EXIST);
        }
        DataWarehouseDatabase resDatabase = this.dwhDatabaseService.getById(resTable.getDatabaseId());
        Set<IColumn> queryTableColumns = this.queryTableColumns(resTable.getTableName(), resDatabase.getDatabaseName());
        DataWarehouseTableDetailVo resTableVo = new DataWarehouseTableDetailVo();
        BeanUtils.copyProperties(resTable, resTableVo);
        resTableVo.setDatabase(resDatabase);
        resTableVo.setSourceAccess(this.sourceAccessService.getById(resDatabase.getId()));
        resTableVo.setColumns(queryTableColumns);
        return resTableVo;
    }

    private Set<IColumn> queryTableColumns(String tableName, String databaseName) throws Exception {
        Connection trinoConnection = trinoQueryUtils.createTrinoConnection();
        ResultSet columns = trinoConnection.getMetaData().getColumns(trinoQueryUtils.getDatawarehouse(), databaseName, tableName, "%");
        Set<IColumn> columnSet = new HashSet<>();
        while (columns.next()) {
            String columnName = columns.getString("COLUMN_NAME");
            String columnType = columns.getString("TYPE_NAME");
            System.out.println("Column: " + columnName + ", Type: " + columnType);
            columnSet.add(new IColumn().setColumnName(columnName).setColumnType(columnType));
        }
        return columnSet;
    }

    @Override
    public long countTablesByDataType(Integer dataType) {
        return sourceTableMapper.countStructuredTables(dataType);
    }

    @Override
    public long countTablesByHierarchy(Integer hierarchy) {
        LambdaQueryWrapper<DataWarehouseTable> queryWrapper = new LambdaQueryWrapper<DataWarehouseTable>()
                .eq(DataWarehouseTable::getHierarchy, hierarchy);
        return count(queryWrapper);
    }


    /**
     * 获取源数据表树形结构
     * @return
     */
    @Override
    public TreeVo tree() {
        // 把各层的表查出来 逐步封装Vo
        List<DataWarehouseDatabase> sourceDatabases = dwhDatabaseService.list();
        List<DataWarehouseTable> sourceTables = list();
        // 封装源数据库
        List<DatabaseVo> sourceDatabasesVo = new ArrayList<>();
        for (DataWarehouseDatabase sourceDatabase : sourceDatabases) {
            DatabaseVo databaseVo = new DatabaseVo();
            BeanUtils.copyProperties(sourceDatabase, databaseVo);
            List<HierarchyVo> hierarchyVos = new ArrayList<>();
            for (DataWarehouseTable sourceTable : sourceTables) {
                if (Objects.equals(sourceTable.getDatabaseId(), sourceDatabase.getId())) {
                    // 查找或创建 HierarchyVo
                    HierarchyVo hierarchyVo = hierarchyVos.stream()
                            .filter(h -> Objects.equals(h.getHierarchy(), sourceTable.getHierarchy()))
                            .findFirst()
                            .orElseGet(() -> {
                                HierarchyVo newHierarchyVo = new HierarchyVo().setHierarchy(sourceTable.getHierarchy()).setTables(new ArrayList<>());
                                hierarchyVos.add(newHierarchyVo);
                                return newHierarchyVo;
                            });
                    // 创建 TableVo 并添加到 HierarchyVo 中
                    TableVo tableVo = new TableVo();
                    BeanUtils.copyProperties(sourceTable, tableVo);
                    hierarchyVo.getTables().add(tableVo);
                }
            }
            databaseVo.setHierarchys(hierarchyVos);
            sourceDatabasesVo.add(databaseVo);
        }
        if (sourceDatabasesVo.isEmpty()){
            return null;
        }
        return new TreeVo().setDatabaseType(ParamConstants.SOURCE_DATABASE).setDatabases(sourceDatabasesVo);
    }


    /**
     * 查询表 可根据 数据库id 数据库名称 层级
     * @param databaseId
     * @param databaseName
     * @param hierarchy
     * @return
     */
    @Override
    public ResponseResult getTable(Long databaseId, String databaseName, Integer hierarchy) {
        List<TableVo> tables =sourceTableMapper.getTable(databaseId, databaseName, hierarchy);
        if(tables.isEmpty()){
            return ResponseResult.fail(CommonConstants.RECORD_NOT_EXIST);
        }
        return ResponseResult.success(tables);
    }



    /**
     * 查询详细信息(分页)
     * @param tablePageDto
     * @return
     */
    @Override
    public ResponseResult detilPage(TablePageDto tablePageDto) {
        if(tablePageDto == null){
            return ResponseResult.fail(CommonConstants.PARAM_IS_NULL);
        }
        tablePageDto.checkParam();
        IPage<DataWarehouseTable> page = Page.of(tablePageDto.getPage(), tablePageDto.getSize());
        LambdaQueryWrapper<DataWarehouseTable> queryWrapper = new LambdaQueryWrapper<>();
        // 因为需要根据databaseName查 所以要先查这个信息 还要分页
        // 这里构造查询条件
        if(tablePageDto.getDatabaseName()!= null){
            Long databaseId = databaseAllService.getIdByDatabaseName(tablePageDto.getDatabaseName());
            if(databaseId == null || !databaseId.equals(tablePageDto.getDatabaseId())){
                return ResponseResult.fail(CommonConstants.PARAM_IS_ERROR);
            }
            queryWrapper.eq(DataWarehouseTable::getDatabaseId, databaseId);
        }else if(tablePageDto.getDatabaseId() != null){
            queryWrapper.eq(DataWarehouseTable::getDatabaseId, tablePageDto.getDatabaseId());
        }
        if(tablePageDto.getHierarchy() != null){
            queryWrapper.eq(DataWarehouseTable::getHierarchy, tablePageDto.getHierarchy());
        }
        // 查询
        if(!queryWrapper.getExpression().getNormal().isEmpty()){
            page = page(page, queryWrapper);
        }
        else{
            page = page(page);
        }
        // 没查到信息
        if(page.getRecords().isEmpty()){
            return ResponseResult.fail(CommonConstants.RECORD_NOT_EXIST);
        }
        // 封装返回
        List<TableDetilVo> tableDetilVos = databaseAllService.createTableDetilVoListForSource(page.getRecords());
        return PageResponseResult.success((int) page.getCurrent(), (int) page.getPages(), (int) page.getSize(), (int) page.getTotal() ,tableDetilVos);
    }

    @Override
    public DataWarehouseTable queryByNameAndDatabaseId(String sourceTable, String targetTable, Long id) {
        // 查询是否存在
        return this.lambdaQuery()
                .eq(DataWarehouseTable::getTableName, targetTable)
                .eq(sourceTable != null, DataWarehouseTable::getSourceTableName, sourceTable)
                .eq(DataWarehouseTable::getDatabaseId, id)
                .last("limit 1")
                .one();
    }

    @Override
    public void deleteByTargetDatabaseName(String dbName) {
        // TODO: 还没有实现
        throw new UnsupportedOperationException("Unimplemented method 'deleteByTargetDatabaseName'");
    }

    @Override
    // hierarchy: 0-ODS 1-DIM 2-DWD
    public DataWarehouseTable createTable(Long targetDatabaseId, String targetTable, Long sourceTableId, Integer hierarchy, String description) {
        DataWarehouseTable newTable = new DataWarehouseTable()
            .setDatabaseId(targetDatabaseId)
            .setTableName(targetTable)
            .setSourceTableId(sourceTableId)
            .setHierarchy(hierarchy)
            .setDescription(description)
                .setCreatedBy(OperatorUtil.getCurrentOperator());
        if(sourceTableId != null && hierarchy != 2){
            DataWarehouseTable sourceTable = getById(sourceTableId);
            if(sourceTable == null){
                throw new IllegalArgumentException("sourceTableId无效，未找到对应的DataWarehouseTable");
            }
            newTable.setSourceTableName(sourceTable.getTableName());
        }
        this.save(newTable);
        return this.getById(newTable.getId());
    }

    @Override
    public DataWarehouseTable createTable(Long targetDatabaseId, String targetTable, Long sourceTableId, Integer hierarchy, String description, Set<IColumn> columns) {
        DataWarehouseTable newTable = new DataWarehouseTable()
            .setDatabaseId(targetDatabaseId)
            .setTableName(targetTable)
            .setSourceTableId(sourceTableId)
            .setHierarchy(hierarchy)
            .setDescription(description);
        DataWarehouseDatabase targetDatabase = this.dwhDatabaseService.getById(targetDatabaseId);
        this.paimonBaseDao.createPaimonTable(targetDatabase.getDatabaseName(), targetTable, columns);
        return newTable;
    }

    @Override
    public void removeByDatabase(DataWarehouseDatabase byDatabaseName) {
        this.lambdaQuery()
            .eq(DataWarehouseTable::getDatabaseId, byDatabaseName.getId())
            .list()
            .forEach(item -> this.removeById(item.getId()));
    }

    @Override
    public void addTag(DataWarehouseTable table, String tag) {
        DataWarehouseDatabase database = this.dwhDatabaseService.getById(table.getDatabaseId());
        this.paimonBaseDao.createTagAtLatestSnapshot(database.getDatabaseName(), table.getTableName(), tag);
    }

    @Override
    public void deleteTag(DataWarehouseTable table, String tag) {
        DataWarehouseDatabase database = this.dwhDatabaseService.getById(table.getDatabaseId());
        this.paimonBaseDao.deleteTag(database.getDatabaseName(), table.getTableName(), tag);
    }

    @Override
    public Set<PaimonTag> fetchTags(DataWarehouseTable table) {
        DataWarehouseDatabase database = this.dwhDatabaseService.getById(table.getDatabaseId());
        return this.paimonBaseDao.showTags(database.getDatabaseName(), table.getTableName());
    }

    @Override
    public void rollBackToTag(DataWarehouseTable table, String tag) {
        DataWarehouseDatabase database = this.dwhDatabaseService.getById(table.getDatabaseId());
        this.paimonBaseDao.rollBackToTag(database.getDatabaseName(), table.getTableName(), tag);
    }

    @Override
    public Set<DataWarehouseTable> traceUp(DataWarehouseTable table) {
        Set<DataWarehouseTable> tables = new HashSet<>();
        this.tableRelationService.lambdaQuery()
            .eq(TableRelation::getTargetTableId, table.getId())
            .list()
            .forEach(item -> {
                DataWarehouseTable sourceTable = this.getById(item.getSourceTableId());
                if(sourceTable != null){
                    table.setSourceTableId(sourceTable.getId());
                    tables.add(sourceTable);
                }
            });
        return tables;
    }

    @Override
    public Set<DataWarehouseTable> traceDown(DataWarehouseTable table) {
        Set<DataWarehouseTable> tables = new HashSet<>();
        this.tableRelationService.lambdaQuery()
            .eq(TableRelation::getSourceTableId, table.getId())
            .list()
            .forEach(item -> {
                DataWarehouseTable targetTable = this.getById(item.getTargetTableId());
                if(targetTable != null){
                    table.setSourceTableId(targetTable.getId());
                    tables.add(targetTable);
                }
            });
        return tables;
    }

    @Override
    public Boolean checkTableNameExist(Long id, String saveTableName) {
        String databaseName = dwhDatabaseService.getById(id).getDatabaseName();
        return this.paimonBaseDao.checkTableExists(databaseName, saveTableName);
    }

    @Override
    @Transactional
    public void deleteTable(DataWarehouseTable table) {
        if(table == null){
            throw new IllegalArgumentException("tableId无效，未找到对应的DataWarehouseTable");
        }
        // 即使湖仓中不存在该表，也继续执行后续操作以清理数据库记录
        DataWarehouseDatabase database = this.dwhDatabaseService.getById(table.getDatabaseId());
        if(this.checkTableNameExist(table.getDatabaseId(), table.getTableName()) == false) {
            // 记录警告日志，但不中断执行
            log.warn("湖仓中不存在该表: {}.{}，但仍将继续清理数据库记录",
                    database.getDatabaseName(),
                    table.getTableName());
        }
        this.removeById(table.getId());

        // TODO: 还要删除血缘关系
        List<TableRelation> relations = this.tableRelationService.lambdaQuery()
                .eq(TableRelation::getSourceTableId, table.getId())
                .or()
                .eq(TableRelation::getTargetTableId, table.getId())
                .list();
        if (!relations.isEmpty()) {
            this.tableRelationService.removeBatchByIds(
                    relations.stream().map(item -> item.getId()).toList()
            );
        }
        // 只有当表在湖仓中确实存在时才尝试删除湖仓中的表
        if(this.checkTableNameExist(table.getDatabaseId(), table.getTableName())) {
            this.paimonBaseDao.deleteTable(database.getDatabaseName(), table.getTableName());
        }
    }

    @Override
    public void deleteTable(Long tableId) {
        DataWarehouseTable table = this.getById(tableId);
        if (table != null) {
            this.deleteTable(table);
        }
    }

    @Override
    public DataWarehouseTable getPaimonTableByDatabaseAndTableName(String database, String table) {
        DataWarehouseDatabase databaseId = dwhDatabaseService.lambdaQuery()
            .eq(DataWarehouseDatabase::getDatabaseName, database)
            .eq(DataWarehouseDatabase::getSourceType, "paimon")
            .one();
        if(databaseId == null){
            return null;
        }
        return this.lambdaQuery()
            .eq(DataWarehouseTable::getDatabaseId, databaseId)
            .eq(DataWarehouseTable::getTableName, table)
            .one();
    }
    
}
